package com.example.kafka2.controller;


import com.example.kafka2.model.Product;
import com.example.kafka2.model.ProductSource;
import com.example.kafka2.util.KafkaUtils;
import com.google.gson.Gson;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Random;
//单线程
@RestController
@RequestMapping("/poc")
public class ProductController {

    private static boolean isrun = false;
    private static int incr = 0;

    @RequestMapping("start")
    public Boolean start(Integer num,String topica,String topicb,Integer sleep,Integer fl,Integer pro,Integer sou){
        try {
            isrun = true;
            Random rand = new Random();
            Gson gson = new Gson();
            if(num != null && num > 0){
                incr= num;
            }
            while (isrun){
                if(sleep != null && sleep > 0){
                    Thread.sleep(sleep*1000);
                }
                incr++;
                int merchant_fullname = rand.nextInt(fl)+1;  // 获取一个随机整数
                int product_id = rand.nextInt(pro)+1;  // 获取一个随机整数

                int score = rand.nextInt(51)+50;
                long ts =System.currentTimeMillis();
                Product product = new Product("分类"+merchant_fullname,"产品"+product_id,
                        "product"+product_id,"v1.1",incr,
                        score,""+product_id,ts);
                int soure_id = rand.nextInt(sou)+1;  // 获取一个随机整数
                ProductSource productSource = new ProductSource(incr,""+soure_id,ts);
                RecordMetadata metadata1 = KafkaUtils.sendmsg(topica, gson.toJson(product));

                System.out.println("当前生成数据roductNo是："+incr+"成功发送product消息，主题：" + metadata1.topic()+"消息的分区：" + metadata1.partition()+"消息的偏移量：" + metadata1.offset());
                RecordMetadata metadata2 = KafkaUtils.sendmsg(topicb, gson.toJson(productSource));
                System.out.println("当前生成数据productNo是："+incr+"成功发送productSource消息，主题：" + metadata2.topic()+"消息的分区：" + metadata2.partition()+"消息的偏移量：" + metadata2.offset());
            }
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @RequestMapping("stop")
    public Boolean stop(){
        isrun = false;
        return true;
    }
}
